package com.kongyz.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * zookeeper 连接管理
 */
public class ConnectManage {
    private static final Logger logger= LoggerFactory.getLogger(ConnectManage.class);
    private volatile static ConnectManage connectManage;

    private EventLoopGroup eventLoopGroup=new NioEventLoopGroup(4);
    private static ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(16,16,600L, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(65536));

    /**
     * CopyOnWriteArrayList：写时复制的容器，
     * 该容器可进行并发的读，而不需要加锁。也是一种读写分离的思想
     */
    private CopyOnWriteArrayList<RpcClientHandler> connectedHandlers=new CopyOnWriteArrayList<RpcClientHandler>();
    private Map<InetSocketAddress,RpcClientHandler> connectedServerNodes=new ConcurrentHashMap<InetSocketAddress, RpcClientHandler>();

    private ReentrantLock lock=new ReentrantLock();
    private Condition connected=lock.newCondition();
    private long connectTimeoutMillis=6000;
    private AtomicInteger roundRobin=new AtomicInteger(0);
    private volatile boolean isRuning=true;

    private ConnectManage() {
    }

    /**
     * 双否定单例模式
     * @return
     */
    public static ConnectManage getInstance(){
        if(connectManage==null){
            synchronized (ConnectManage.class){
                if(connectManage==null){
                    connectManage=new ConnectManage();
                }
            }
        }
        return connectManage;
    }

    /**
     * 更新服务连接
     * @param allServerAddress
     */
    public void updateConnectedServer(List<String> allServerAddress){
        if(allServerAddress!=null){
            if(allServerAddress.size()>0){
                HashSet<InetSocketAddress> newAllServerNodeSet=new HashSet<>();
                for (int i=0;i<allServerAddress.size();i++){
                    String[] array=allServerAddress.get(i).split(":");
                    if(array.length==2){
                        String host=array[0];
                        int port=Integer.parseInt(array[1]);
                        final InetSocketAddress remotePeer=new InetSocketAddress(host,port);
                        newAllServerNodeSet.add(remotePeer);
                    }
                }

                //添加新的服务连接
                for (final InetSocketAddress serverNodeAddress:newAllServerNodeSet){
                    if(!connectedServerNodes.keySet().contains(serverNodeAddress)){
                        connectServerNode(serverNodeAddress);
                    }
                }
                //关闭并移除无效的连接
                for (int i=0;i<connectedHandlers.size();i++){
                    RpcClientHandler connectedServerHandler=connectedHandlers.get(i);
                    SocketAddress remotePeer=connectedServerHandler.getRemotePeer();
                    if(!newAllServerNodeSet.contains(remotePeer)){
                        logger.info("删除无效的服务节点 ： "+remotePeer);
                        RpcClientHandler handler=connectedServerNodes.get(remotePeer);
                        if(handler!=null){
                            handler.close();
                        }
                        connectedServerNodes.remove(remotePeer);
                        connectedHandlers.remove(connectedServerHandler);
                    }
                }
            }else {
                logger.error("没有可用的服务节点，所有服务节点均已关闭");
                for (final RpcClientHandler connectedServerhandler:connectedHandlers){
                    SocketAddress remotePeer=connectedServerhandler.getRemotePeer();
                    RpcClientHandler handler=connectedServerNodes.get(remotePeer);
                    handler.close();
                    connectedServerNodes.remove(connectedServerhandler);
                }
                connectedHandlers.clear();
            }
        }
    }

    /**
     * 连接
     * @param remotePeer
     */
    private void connectServerNode(final InetSocketAddress remotePeer){
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("远程服务地址："+remotePeer);
                Bootstrap b=new Bootstrap();
                b.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        .handler(new RpcClientInitializer());

                final ChannelFuture channelFuture=b.connect(remotePeer);
                channelFuture.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(final ChannelFuture future) throws Exception {
                        if(future.isSuccess()){
                            logger.debug("远程服务器链接成功！远程地址："+remotePeer);
                            RpcClientHandler handler=channelFuture.channel().pipeline().get(RpcClientHandler.class);
                            addHandler(handler);
                        }
                    }
                });
            }
        });
    }

    private void addHandler(RpcClientHandler handler){
        connectedHandlers.add(handler);
        InetSocketAddress remoteAddress=(InetSocketAddress)handler.getChannel().remoteAddress();
        connectedServerNodes.put(remoteAddress,handler);
        signaAvailableHandler();
    }

    /**
     * 发起所有通知
     */
    private void signaAvailableHandler(){
        lock.lock();
        try {
            connected.signalAll();
        }finally {
            lock.unlock();
        }
    }

    /**
     * 发起等待
     * @return
     * @throws InterruptedException
     */
    private boolean waitingForHandler() throws InterruptedException {
        lock.lock();
        try {
            return connected.await(this.connectTimeoutMillis,TimeUnit.MILLISECONDS);
        }finally {
            lock.unlock();
        }
    }


    public RpcClientHandler chooseHandler(){
        int size=connectedHandlers.size();
        while (isRuning&&size<=0){
            try {
                boolean available=waitingForHandler();
                if(available){
                    size=connectedHandlers.size();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                logger.error("没有等待连接节点！",e);
                throw  new RuntimeException("不能链接服务器",e);
            }
        }
        int index=(roundRobin.getAndAdd(1)+size)%size;
        return connectedHandlers.get(index);
    }



    public void stop(){
        isRuning =false;
        for (int i=0;i<connectedHandlers.size();i++){
            RpcClientHandler connectedServerHandler=connectedHandlers.get(i);
            connectedServerHandler.close();
        }
        signaAvailableHandler();
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }









}
